package com.hxzy.rocketmqproducer.controller;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * 向服务器发送消息（生产者）
 */
@Log4j2
@RestController
public class SendMessageController {

    /**
     * 生产者
     */
    @Autowired
    protected  DefaultMQProducer  defaultMQProducer;


    @GetMapping(value = "/send")
    public String send(String msg) throws MQClientException, UnsupportedEncodingException, MQBrokerException, RemotingException, InterruptedException {

        // 创建消息，并指定Topic，Tag和消息体  (主题)
        Message message = new Message("HxzyTopic" /* Topic */,
                    "TagA" /* Tag */,
                msg.getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );

        // 发送消息到一个Broker  (同步发送)
        SendResult sendResult = defaultMQProducer.send(message);
        // 通过sendResult返回消息是否成功送达
        System.out.printf("%s%n", sendResult);


        return "发送消息成功";
    }

    @GetMapping(value = "/sendtag")
    public String sendTag(String msg,String tags) throws MQClientException, UnsupportedEncodingException, MQBrokerException, RemotingException, InterruptedException {

        // 创建消息，并指定Topic，Tag和消息体  (主题)
        Message message = new Message("HxzyTopic" /* Topic */,
                tags /* Tag */,
                msg.getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        );

        // 发送消息到一个Broker  (同步发送)
        SendResult sendResult = defaultMQProducer.send(message);
        // 通过sendResult返回消息是否成功送达
        System.out.printf("%s%n", sendResult);


        return "发送消息成功";
    }

    @GetMapping(value = "/send_json")
    public String sendJson(String msg) throws MQClientException, UnsupportedEncodingException, MQBrokerException, RemotingException, InterruptedException, JsonProcessingException {

        Map<String,Object> map=new HashMap<>();
        map.put("code",0);
        map.put("message",msg);
        map.put("createDate",new Date());

        //Jsonson2序列化
        ObjectMapper om = new ObjectMapper();
        om.setSerializationInclusion(JsonInclude.Include.NON_NULL);
        String jsonString=om.writer().withDefaultPrettyPrinter().writeValueAsString(map);



        // 创建消息，并指定Topic，Tag和消息体  (主题)
        Message message = new Message("HxzyTopic" /* Topic */,
                "TagA" /* Tag */,
                jsonString.getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        );

        // 发送消息到一个Broker  (同步发送)
        SendResult sendResult = defaultMQProducer.send(message);
        // 通过sendResult返回消息是否成功送达
        System.out.printf("%s%n", sendResult);


        return "发送消息成功";
    }

    /**
     * 异步发送 (大批量发送操作   1种之内有大量异步请求操作)
     * @param msg
     * @return
     * @throws MQClientException
     * @throws UnsupportedEncodingException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    @GetMapping(value = "/send_asnyc")
    public String sendAsnyc(String msg) throws MQClientException, UnsupportedEncodingException, MQBrokerException, RemotingException, InterruptedException {

        // 创建消息，并指定Topic，Tag和消息体  (主题)
        Message message = new Message("HxzyTopic" /* Topic */,
                "TagB" /* Tag */,
                msg.getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        );
        // 发送消息到一个Broker  (同步发送)
        defaultMQProducer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 通过sendResult返回消息是否成功送达
                System.out.println("异步消息发送成功---");
                System.out.printf("%s%n", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.debug(throwable.getMessage());
            }
        });




        return "发送消息成功";
    }
}
